package com.shujia.dw

import java.text.SimpleDateFormat

import com.google.gson.Gson
import com.shujia.bean.CaseClass.{MergeLocation, StayPoint}
import com.shujia.dw.MakeMergeLocaltionApp.LOGGER
import com.shujia.exception.TourException
import com.shujia.util.{Config, SparkTool}
import org.apache.spark.sql.{SQLContext, SaveMode}
import com.shujia.grld.Grid

object MakeStayPointApp extends SparkTool {
  /**
    * 编写spark业务逻辑
    *
    */
  override def run(args: Array[String]): Unit = {


    val mergeLocationPath = s"${Config.getString("mergelocation.path")}day_id=$day_id"
    val staypointPath = s"${Config.getString("staypoint.path")}day_id=$day_id"


    LOGGER.info(s"融合表输入路径:$mergeLocationPath")
    LOGGER.info(s"停留表输出路径:$staypointPath")

    val sQLContext = new SQLContext(sc)
    import sQLContext.implicits._

    //读取融合表数据
    val mergeDF = sQLContext.read.parquet(mergeLocationPath)


    /**
      * 将同一个人在同一个网格的数据合并成一条数据
      *
      */

    val caseRDD = mergeDF
      .toJSON
      //将数据转换成自定义类的对象
      .map(json => {
      val gson = new Gson()
      gson.fromJson(json, classOf[MergeLocation])
    })

    val stayPointRDD = caseRDD.map(merge => {
      val key = merge.mdn + "_" + merge.grid_id + "_" + merge.county_id
      (key, merge.start_time)
    }).groupByKey()
      .map(kv => {
        val split = kv._1.split("_")
        val mdn = split(0)
        val grid_id = split(1)
        val county_id = split(2)

        //通过时间进行排序
        val sortData = kv
          ._2
          .toList
          .sortBy(line => line.split(",")(1).toLong)


        //第一条数据
        val head = sortData.head
        //最后一条数据
        val last = sortData.last

        val simpleDateFormat = new SimpleDateFormat("yyyyMMddhhmmss")

        val lastDate = simpleDateFormat.parse(last.split(",")(0))
        val headDate = simpleDateFormat.parse(head.split(",")(1))

        //计算用户在网格里面的停留时间
        val duration = Math.abs((lastDate.getTime - headDate.getTime) / 60000)


        //获取网格点的经纬度
        val point = Grid.getCenter(grid_id.toLong)

        val longi = point.x
        val lati = point.y


        //构建停留表对象
        StayPoint(mdn, longi.toString, lati.toString, grid_id, county_id, duration.toString, head.split(",")(1), last.split(",")(0))
      })

    stayPointRDD
      .toDF()
      .write
      .mode(SaveMode.Overwrite)
      .parquet(staypointPath)
  }

  /**
    * 初始化方法，在子类设置spark运行时需要的参数
    * conf.setMaster("local")
    */
  override def init(): Unit = {

  }
}
